package com.minelx.datamicroscope.core.query;

import com.minelx.datamicroscope.core.condition.KeyOperandCondition;
import com.minelx.datamicroscope.core.condition.RawCondition;
import com.minelx.datamicroscope.core.index.IndexFactory;
import com.minelx.datamicroscope.core.values.CommonField;
import com.minelx.hellopandas.core.DataFrame;
import com.minelx.hellopandas.core.DataFrameDescription;
import io.gitee.minelx.filesystem.File;

import java.util.List;
import java.util.Map;

import static java.util.stream.Collectors.toList;

class QueryEmitter {

    private final Resources resources;

    private final IndexFactory indexFactory;

    private final RawConditionFactory rawConditionFactory;

    QueryEmitter(Resources resources,
                 IndexFactory indexFactory,
                 RawConditionFactory rawConditionFactory) {
        this.resources = resources;
        this.indexFactory = indexFactory;
        this.rawConditionFactory = rawConditionFactory;
    }

    void collectInContainer(Map<String, CommonField> connection,
                            List<KeyOperandCondition> extraConditions,
                            String resourceName,
                            List<AliasedKey> collectedFields) {
        List<RawCondition> rawConditions = transformToRawConditions(connection, extraConditions);

        List<AliasedKey> connectedKeys = connectionToKeys(connection);
        CollectingResource collectingResource = CollectingResource.create(collectedFields);
        File resourceFile = resources.createResource(resourceName, connectedKeys, collectingResource.getCollectedFields());
        indexFactory.createIndex(connectedKeys, rawConditions, collectingResource, resourceFile).list();
    }

    private List<RawCondition> transformToRawConditions(Map<String, CommonField> connection,
                                                        List<KeyOperandCondition> extraConditions) {
        List<RawCondition> result = extraToRawConditions(extraConditions);
        // add connecting condition
        RawCondition connectionConditions = connectionToRawConditions(connection);
        result.add(connectionConditions);
        return result;
    }

    private RawCondition connectionToRawConditions(Map<String, CommonField> connection) {
        List<AliasedKey> keys = connectionToKeys(connection);
        DataFrame valuesDataFrame = resources.identify(keys);
        DataFrameDescription valuesDescription = valuesDataFrame.description();
        return rawConditionFactory.rawConditionFromDataFrameSource(keys, valuesDescription);
    }

    private List<RawCondition> extraToRawConditions(List<KeyOperandCondition> extraConditions) {
        return extraConditions.stream()
                .map(rawConditionFactory::extraToRawCondition)
                .collect(toList());
    }

    private List<AliasedKey> connectionToKeys(Map<String, CommonField> connection) {
        return connection.entrySet().stream()
                .map(stringFieldLocateEntry -> AliasedKey.create(stringFieldLocateEntry.getKey(), stringFieldLocateEntry.getValue()))
                .collect(toList());
    }
}
